%%%-------------------------------------------------------------------
%%% File    : bqueue.erl
%%% Author  :  <vjache>
%%% Description : 
%%%
%%% Created : 13 Jul 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(bqueue).

-include_lib("eunit/include/eunit.hrl").

-include("../include/pipe.hrl").
-include("../include/iter.hrl").

-export([create/1, 
		 destroy/2, 
		 new_pipe/1, 
		 new_iter/1,
		 new_pipe_one2many_pattern/2,
		 init/1, 
		 test_prod_cons_bulk/0]).

-record(state,{owner_pid=self(),
			   pending_push=queue:new(),
			   push_size=0,
			   pending_pop=queue:new(),
			   pop_size=0,
			   queue=queue:new(),
			   qsize=0,
			   max_qsize,
			   is_destroying=false}).

%% -define(ECHO(Format,Data),error_logger:info_msg(Format,Data)).
-define(ECHO(Format,Data),noop(Data)).


create(MaxSize) ->
	Pid=spawn_link(bqueue, init, [#state{max_qsize=MaxSize}]),
%% 	timer:sleep(100),
%% 	true=is_process_alive(Pid),
	Pid.
destroy(QPid, IsForced) when is_boolean(IsForced) ->
	case is_process_alive(QPid) of
		false ->
			throw({no_such_queue, QPid});
		true ->
			ReplyPid=self(),
			QPid ! {destroy, ReplyPid,IsForced},
			receive
				{destroy_ok, QPid} ->
					QRef=erlang:monitor(process, QPid),
					receive
						{'DOWN', QRef, process, QPid, normal} ->
							ok;
						{'DOWN', QRef, process, QPid, noproc} ->
							ok;
						{'DOWN', QRef, process, QPid, noconnection} ->
							ok
					end;
				{destroy_err, QPid, not_a_owner} ->
					throw({not_a_owner,self()})
			end
	end.

open_sess(QPid) ->
	QRef=erlang:monitor(process, QPid),
	{QRef,QPid,self()}.

close_sess({QRef,QPid,SelfPid}) ->
	case self() of
		SelfPid ->
			erlang:demonitor(QRef),
			receive
				{'DOWN', QRef, process, QPid, _} ->
					ok
				after 0 ->
					ok
			end;
		Pid ->
			throw({not_a_client,Pid})
	end.
	
push({QRef,QPid,SelfPid}, Item) ->
	case self() of
		SelfPid ->
			ReqId=make_ref(),
			QPid ! {push, ReqId, SelfPid, Item},
			receive
				{QPid, ReqId, ok} ->
					ok;
				{'DOWN', QRef, process, QPid, normal} ->
					throw({queue_unexpected_down,QPid});
				{'DOWN', QRef, process, QPid, noproc} ->
					throw({no_such_queue, QPid});
				{'DOWN', QRef, process, QPid, noconnection} ->
					throw({remote_queue_noconnection,QPid})
			end;
		Pid ->
			throw({not_a_client,Pid})
	end.

pop({QRef,QPid,SelfPid}) ->
	case self() of
		SelfPid ->
			ReqId=make_ref(),
			QPid ! {pop, ReqId, SelfPid},
			receive
				{QPid,ReqId,'$EOQ'} ->
					'$EOQ';
				{QPid, ReqId, Item} ->
					Item;
				{'DOWN', QRef, process, QPid, normal} ->
					'$EOQ';
				{'DOWN', QRef, process, QPid, noproc} ->
					'$EOQ';
				{'DOWN', QRef, process, QPid, noconnection} ->
					throw({remote_queue_noconnection,QPid})
			end;
		Pid ->
			throw({not_a_client,Pid})
	end.



init(State) ->
	loop(State).

loop({allowed_destroy,State}) ->
	case queue:len(PopQueue=State#state.pending_pop) of
		0 ->
			ok;
		_ ->
			lists:foreach(
			  fun({pop, ReqId, ClientPid}) ->
					  ClientPid ! {self(), ReqId, '$EOQ'}
			  end, queue:to_list(PopQueue))
	end,
	State#state.owner_pid ! {destroy_ok, self()};
loop(State=#state{
				  owner_pid=OwnerPid,
				  pending_push=PushQueue,
				  push_size=PushSize,
				  pending_pop=PopQueue,
				  pop_size=PopSize,
				  queue=_Queue,
				  qsize=_QueueSize,
				  max_qsize=_QueueMaxSize,
				  is_destroying=IsDestr}) ->
	
	receive
		%% Short circuits optimization begin
		%% push & (in_q=0,q=0,out_q!=0)
		{push, ReqId1, ReplyPid1, Item}=_Msg 
		  when PushSize==0 andalso qsize==0 andalso pop_size>0  ->
			?ECHO("S/C",[]),
			throw(sc),
			{{value, {pop, ReqId, ReplyPid}}, PopQ1}=queue:out(PopQueue),
			ReplyPid ! {self(), ReqId, Item},
			ReplyPid1 ! {self(), ReqId1, ok},
			State1=State#state{pending_pop=PopQ1,
							   pop_size=PopSize-1},
			case IsDestr of 
				true ->
					loop(try_pump(State1));
				false ->
					loop(State1)
			end;
		%% pop & (in_q=_,q!=0,out_q=0)
		%% Short circuits optimization end
		{pop, _ReqId, _ClientPid}=Msg ->
			State1=State#state{pending_pop=queue:in(Msg, PopQueue),pop_size=PopSize+1},
			loop(try_pump(State1));
		{push, _ReqId, _ClientPid, _Item}=Msg ->
			State1=State#state{pending_push=queue:in(Msg, PushQueue),push_size=PushSize+1},
			loop(try_pump(State1));
		{destroy, OwnerPid, IsForced} -> %% Owner close
%% 			?ECHO("ON DESTROY~n",[]),
			case try_pump(flush_mbox(State#state{is_destroying=true})) of
				State1 when IsForced andalso
							is_record(State1, state)-> %% Forced destroy !
					error_logger:warning_report(
					  [{slogan,{queue_not_empty, forced_destroy}},
					   {qsize,State1#state.qsize},
					   {pending_push_size,State1#state.push_size},
					   {pending_pop_size,State1#state.pop_size}]),
					loop({allowed_destroy,State1});
				State1 -> %% Proceed
					loop(State1)
			end;
		{destroy, ClientPid} ->
			ClientPid ! {destroy_err, self(), not_a_owner},
			loop(State)
	end.

flush_mbox(State=#state{
				  owner_pid=_OwnerPid,
				  pending_push=PushQueue,
				  push_size=PushSize,
				  pending_pop=PopQueue,
				  pop_size=PopSize,
				  queue=_Queue,
				  qsize=_QueueSize,
				  max_qsize=_QueueMaxSize,
				  is_destroying=_}) ->
	receive
		{pop, _ClientRef, _ClientPid}=Msg ->
			State1=State#state{pending_pop=queue:in(Msg, PopQueue),pop_size=PopSize+1},
			flush_mbox(State1);
		{push, _ClientRef, _ClientPid, _Item}=Msg ->
			State1=State#state{pending_push=queue:in(Msg, PushQueue),push_size=PushSize+1},
			flush_mbox(State1)
		after 0 ->
			State
	end.

try_pump(State) %% Pop
  when State#state.pop_size > 0 andalso State#state.qsize > 0 ->
	{{value, Item}, ItemQ1}=queue:out(State#state.queue),
	{{value, {pop, ReqId, ReplyPid}}, PopQ1}=queue:out(State#state.pending_pop),
	ReplyPid ! {self(), ReqId, Item},
	try_pump(State#state{
				queue=ItemQ1,
				qsize=State#state.qsize-1,
				pending_pop=PopQ1,
				pop_size=State#state.pop_size-1
				});
try_pump(State) %% Push
  when State#state.push_size>0 andalso 
	   State#state.qsize<State#state.max_qsize ->
	{{value, {push, ReqId, ReplyPid, Item}}, PushQ1}=queue:out(State#state.pending_push),
	ItemQ1=queue:in(Item, State#state.queue),
	ReplyPid ! {self(), ReqId, ok},
	try_pump(State#state{
				queue=ItemQ1,
				qsize=State#state.qsize+1,
				pending_push=PushQ1,
				push_size=State#state.push_size-1
				});
try_pump(State) %% Close
  when State#state.is_destroying andalso
  	   State#state.push_size==0 andalso
	   State#state.qsize==0 ->
	{allowed_destroy,State};	
try_pump(State) ->
%% 	?ECHO("State noop ~p~n",[State]),
	State.


new_pipe(QPid) ->
	pipe:simple_immutable(
	  fun()->
			  open_sess(QPid)
	  end, 
	  fun(QSess,Item)->
			  push(QSess,Item)
	  end, 
	  fun(QSess)->
			  close_sess(QSess)
	  end).

new_pipe_one2many_pattern(QSize,CosumersFactoryFun) 
  when is_integer(QSize) andalso
	   is_function(CosumersFactoryFun, 1)->
	pipe:simple_immutable(
	  fun()->
			  QPid=create(QSize),
			  QSess=open_sess(QPid),
			  CosumersFactoryFun(new_iter(QPid)),
			  {QSess,QPid}
	  end, 
	  fun({QSess,_QPid},Item)->
			  push(QSess,Item)
	  end, 
	  fun({QSess,QPid})->
			  close_sess(QSess),
			  destroy(QPid, false)
	  end).



new_iter(QPid) ->
	iter:simple_immutable(
	  fun()->
			  open_sess(QPid)
	  end, 
	  fun(QSess)->
			  case pop(QSess) of
				  '$EOQ'->
					  ?ECHO("EOF",[]),
					  ?EOF;
				  Item ->
					  ?ECHO("ITEM ~p~n",[Item]),
					  Item
			  end
	  end, 
	  fun(QSess)->
			  close_sess(QSess)
	  end).
%%======================================================================================
%% eunit tests
%%======================================================================================

push_test() ->
	TestData=lists:seq(1, 10),
	QPid=create(10),
	?ECHO("CREATED~n",[]),
	QSess=open_sess(QPid),
	?ECHO("OPENED~n",[]),
	Results=[push(QSess,Item)||Item<-TestData],
	?ECHO("PUSHED~n",[]),
	?assertEqual([ok||_<-TestData],Results),
	?ECHO("ASSERTED~n",[]),
	close_sess(QSess),
	?ECHO("CLOSED~n",[]),
	destroy(QPid,true),
	?ECHO("DESTROYED~n",[]),
	ok.
pop_test() ->
	QPid=create(10),
	?ECHO("CREATED~n",[]),
	Consumers=[a,b,c,d],
	ConsPids=[spawn_consumer(QPid,Cons)||Cons<-Consumers],
	?ECHO("CONSUMERS STARTED~n",[]),
	timer:sleep(100),
	?assertEqual([true||_ConsPid<-ConsPids],[is_process_alive(ConsPid)||ConsPid<-ConsPids]),
	?ECHO("ASSERTED~n",[]),
	destroy(QPid,false),
	?ECHO("DESTROYED~n",[]),
	timer:sleep(100),
	?assertEqual([false||_ConsPid<-ConsPids],[is_process_alive(ConsPid)||ConsPid<-ConsPids]),
	ok.

push_pop_test() ->
	TestData=lists:seq(1, 10),
	QPid=create(10),
	?ECHO("CREATED~n",[]),
	QSess=open_sess(QPid),
	?ECHO("OPENED~n",[]),
	Results=[push(QSess,Item)||Item<-TestData],
	?ECHO("PUSHED~n",[]),
	?assertEqual([ok||_<-TestData],Results),
	?assertEqual(lists:sort(TestData),lists:sort([pop(QSess)||_<-TestData])),
	close_sess(QSess),
	?ECHO("CLOSED~n",[]),
	destroy(QPid,false),
	?ECHO("DESTROYED~n",[]),
	ok.

prod1_cons4_test() ->
	TestData=lists:seq(1, 400),
	QPid=create(100),
	Consumers=[a,b,c,d],
	[spawn_consumer(QPid,Cons)||Cons<-Consumers],
	QSess=open_sess(QPid),
	Results=[push(QSess,Item)||Item<-TestData],
	close_sess(QSess),
	?assertEqual([ok||_<-TestData],Results),
	destroy(QPid,false),
	?assertEqual(Consumers,[
	 receive
		 Cons ->
			 Cons
		 after
			5000 ->
				err
	 end
	 ||Cons<-Consumers]),
	ok.

test_prod_cons_bulk() ->
	N=40000000000,
	QPid=create(100),
	Consumers="uwfuydgjkasfakcblakjdklfdjldfkmhandlfnaldfwbfekbalkjbflkdjabgkj",
	[spawn_slow_consumer(QPid,Cons)||Cons<-Consumers],
	QSess=open_sess(QPid),
	for(1,N,
		fun(Item)->
				Rem=Item rem 100,
				case Rem of
					0 ->
						io:format("PUSHED ~p at ~p~n",[Item,now()]);
					_ ->
						ok
				end,
				push(QSess,Item)
		end),
	close_sess(QSess),
	destroy(QPid,false),
	?assertEqual(Consumers,[
	 receive
		 Cons ->
			 Cons
		 after
			5000 ->
				err
	 end
	 ||Cons<-Consumers]),
	ok.

prod4_cons1_test() ->
	TestDataList=[lists:seq(N*100,(N+1)*100 )||N<-lists:seq(0,3)],
	QPid=create(10000000000),
	Consumers=[a,b,c],
	Self=self(),
	[spawn_consumer(QPid,Cons)||Cons<-Consumers],
	Prods=[spawn_link(
	   fun()->
			   QSess=open_sess(QPid),
			  lists:foreach(
				fun(Item)->
						push(QSess,Item)
				end, TestData),
			   close_sess(QSess),
			  Self ! {pushed,self()}
	   end)||TestData<-TestDataList],
	[receive
		{pushed,ProdPid} ->
			ok
	end||ProdPid<-Prods],
	?ECHO("PUSHED~n",[]),
	destroy(QPid,false),
	?ECHO("DESTROYED~n",[]),
	?assertEqual(Consumers,[
	 receive
		 Cons ->
			 ?ECHO("CONS FINISH ~p ~n",[Cons]),
			 Cons
	 end
	 ||Cons<-Consumers]),
	ok.

pipe_test() ->
	TestData=lists:seq(1, 10),
	QPid=create(10),
	?ECHO("CREATED~n",[]),
	Pipe=new_pipe(QPid),
	Pump=pump:new(iter:from_list(TestData), Pipe),
	pump:init(Pump),
	?ECHO("PUMPED~n",[]),
	destroy(QPid,true),
	?ECHO("DESTROYED~n",[]),
	ok.
iter_test() ->
	TestData=lists:seq(1, 10),
	QPid=create(10),
	?ECHO("CREATED~n",[]),
	Pipe=new_pipe(QPid),
	Pump=pump:new(iter:from_list(TestData), Pipe),
	pump:init(Pump),
	?ECHO("PUMPED~n",[]),
	spawn_link(
	  fun()->
			  Iter=new_iter(QPid),
			  {Is,Acc}=iter:fold(Iter,
								 fun(Item,AccIn)->
										 [Item|AccIn]
								 end,[]),
			  ?ECHO("ITERATED~n",[]),
			  iter:close(Is),
			  ?ECHO("CLOSED",[]),
			  ?assertEqual(lists:sort(TestData),lists:sort(Acc)),
			  ?ECHO("ASSERTED~n",[])
	  end),
	destroy(QPid,false),
	?ECHO("DESTROYED~n",[]),
	ok.

spawn_consumer(QPid,ConsumerName) ->
	Self=self(),
	spawn_link(
	   fun()->
			  QSess=open_sess(QPid),
			  cons_loop(QSess),
			  close_sess(QSess),
			  Self ! ConsumerName
	   end).
cons_loop(QSess) ->
	case pop(QSess) of
		'$EOQ' ->
			ok;
		Item ->
			?ECHO("~p consumed ~p~n",[self(),Item]),
			cons_loop(QSess)
	end.
spawn_slow_consumer(QPid,ConsumerName) ->
	Self=self(),
	spawn_link(
	   fun()->
			  QSess=open_sess(QPid),
			  slow_cons_loop(QSess),
			  close_sess(QSess),
			  Self ! ConsumerName
	   end).
slow_cons_loop(QSess) ->
	case pop(QSess) of
		'$EOQ' ->
			ok;
		Item ->
			timer:sleep(100),
			?ECHO("~p consumed ~p~n",[self(),Item]),
			slow_cons_loop(QSess)
	end.
noop(_) ->
	ok.
for(I,N,_Fun) when I>N ->
	ok;
for(I,N,Fun) ->
	Fun(I),
	for(I+1,N,Fun).
%% for_fold(0,_Fun,Acc0) ->
%% 	Acc0;
%% for_fold(N,Fun,Acc0) ->
%% 	for_fold(N-1,Fun,Fun(N,Acc0)).